/****************************************************************************
 *
 * Copyright (C) 2003-2008 Los Alamos National Security, LLC
 *                         Packet Analytics Corporation
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License Version 2 as
 * published by the Free Software Foundation.  You may not use, modify or
 * distribute this program under any other version of the GNU General
 * Public License.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 *
 ****************************************************************************/
package nfse.query;

import java.io.*;
import java.net.*;
import java.sql.*;
import java.util.*;

import javax.net.SocketFactory;
import javax.net.ssl.*;

import nfse.*;
import nfse.result.*;

public class StaticQueryProcessor {

    public QueryServer queryServer = null;

    private JDBCDatabase statDB = null;

    private NetFSERecordFactory rec = null;

    public StaticQueryProcessor(QueryServer queryServer, int type) throws InstantiationException, IllegalAccessException, ClassNotFoundException {
        this.queryServer = queryServer;
        rec = NetFSE.getRecordFactory(type);
    }

    public int process(Query query, MetaDatabase mdb) {
        int val = 0;
        Socket resultSocket = null;
        ServerSocket ss = null;
        DataOutputStream resultOut = null;
        SocketFactory sf = SocketFactory.getDefault();
        SocketResult socketResult = null;
        MarshalledResult marshalledResult = null;
        SearchEngine se = null;
        try {
            val = query.validate();
            if (val == Query.QUERY_INVALID) {
                System.err.println("Attempted to process an invalid query: '" + query.toXML() + "'");
                return -1;
            }
            if (val == Query.QUERY_EMPTY) {
                System.out.println("Processed an empty query: '" + query.toXML() + "'");
                return 0;
            }
            int numSubs = query.getNumSubqueries();
            if (numSubs > 0) {
                Query[] subs = query.getSubqueries();
                for (int i = 0; i < numSubs; i++) {
                    subs[i] = subs[i].detach(this.queryServer.nextQueryID());
                    this.process(subs[i], mdb);
                }
                System.gc();
                return 0;
            }

            int queryID = query.getID();
            if (query.getNumDataTypes() == 1) {
                Data datatype = query.getData()[0];
                if (datatype.sensor == -1) {
                    Query[] queries = query.decomposeBySensor();
                    if (queries.length == 0) {

                        return 0;
                    }

                    ResultMarshall rm = null;
                    try {
                        rm = Query
                                .marshallResults(queries, NetFSE.LOCAL_IP_ADDRESS, NetFSE.RESULT_SERVER_PORT, queryID);
                    } catch (QueryException qe) {
                        qe.printStackTrace();
                        return 0;
                    }
                    this.queryServer.getResultServer().addResultMarshall(rm);
                    for (int i = 0; i < queries.length; i++) {
                        queries[i] = queries[i].detach(queryServer.nextQueryID());
                        this.process(queries[i], mdb);
                    }
                    System.gc();
                    return 0;
                }
            }

            Data[] data = query.getData();
            if (data.length != 1) {
                throw (new QueryException("Invalid Data elements specified or failed decomposition."));
            }
            int type = data[0].type;
            int sensor = data[0].sensor;

            DateRange[] dateRanges = query.getDateRanges();
            if ((query.isDistribute())
                    && ((dateRanges.length > 1) || ((dateRanges.length == 1) && (dateRanges[0].isRange())))) {
                Vector<QueryDate> dates = new Vector<QueryDate>();
                Statement s = mdb.createStatement();
                for (int i = 0; i < dateRanges.length; i++) {
                    String sql = "";
                    if (dateRanges[i].isRange()) {
                        sql = "select distinct a.Month, a.Day, a.Year from Static_Indices a, Sensors b where a.Sensor_ID = b.ID and b.ID = "
                                + sensor
                                + " and (a.Min_Time >= "
                                + dateRanges[i].getStartDate().startTS
                                / 1000
                                + " and a.Max_Time <=" + dateRanges[i].getEndDate().endTS / 1000 + ")";
                    } else {
                        sql = "select distinct Month, Day, Year from Static_Indices a, Sensors b where a.Sensor_ID = b.ID and b.ID = "
                                + sensor
                                + " and (a.Min_Time >= "
                                + dateRanges[i].getStartDate().startTS
                                / 1000
                                + " and a.Max_Time <=" + dateRanges[i].getStartDate().endTS / 1000 + ")";
                    }

                    ResultSet rs = s.executeQuery(sql);
                    while (rs.next()) {
                        int m = rs.getInt("Month");
                        int d = rs.getInt("Day");
                        int y = rs.getInt("Year");
                        dates.addElement(new nfse.query.QueryDate(m, d, y));
                    }
                    rs.close();
                    sql = "";

                }
                s.close();

                nfse.query.QueryDate[] sortedDates = new nfse.query.QueryDate[dates.size()];
                for (int i = 0; i < dates.size(); i++) {
                    sortedDates[i] = (nfse.query.QueryDate) dates.elementAt(i);
                }
                Arrays.sort(sortedDates);
                NodeStatus[] queryNodes = NetFSE.getQueryNodeStatus();
                int datesPerNode = 0;
                if (queryNodes.length == 0)
                    datesPerNode = sortedDates.length;
                else
                    datesPerNode = sortedDates.length / queryNodes.length;
                if ((queryNodes.length != 0) && (sortedDates.length % queryNodes.length != 0))
                    datesPerNode++;

                try {
                    socketResult = (SocketResult) query.getResult();
                } catch (ClassCastException cce) {
                    marshalledResult = (MarshalledResult) query.getResult();
                }

                if (datesPerNode == 0) {

                    if (socketResult != null) {
                        resultSocket = sf.createSocket(socketResult.host, socketResult.port);
                        resultOut = new DataOutputStream(resultSocket.getOutputStream());
                    } else if (marshalledResult != null) {
                        resultSocket = sf.createSocket(marshalledResult.host, marshalledResult.port);
                        resultOut = new DataOutputStream(resultSocket.getOutputStream());
                        resultOut.writeInt(marshalledResult.queryID);
                        resultOut.writeInt(NetFSE.RESULT_SERVER_FLAG_DONE);
                        resultOut.writeInt(-100);
                        resultOut.writeInt(0);
                        DataInputStream resultIn = new DataInputStream(resultSocket.getInputStream());
                        resultIn.readInt();
                        resultIn.close();
                    }
                    if (resultOut != null) {
                        resultOut.close();
                        resultSocket.close();
                    }
                    return 0;
                }

                int numNodes = sortedDates.length / datesPerNode;
                if (sortedDates.length % datesPerNode != 0)
                    numNodes++;
                if (datesPerNode == 1)
                    numNodes = sortedDates.length;

                Query[] queries = new Query[numNodes];
                for (int i = 0; i < numNodes; i++) {
                    Query q = ((Query) query.clone()).detach(queryServer.nextQueryID());
                    q.setDate(null);
                    q.setDistribute(false);
                    int min = i * datesPerNode;
                    int max = i * datesPerNode + datesPerNode;
                    Vector<DateRange> temp = new Vector<DateRange>();
                    for (int j = min; j < max; j++) {
                        if (j < sortedDates.length) {
                            temp.addElement(new DateRange(sortedDates[j]));
                        }
                    }
                    DateRange[] dates2 = DateRange.consolidate(temp);

                    for (int j = 0; j < dates2.length; j++) {
                        q.addDateRange(dates2[j]);
                    }
                    queries[i] = q;
                }
                if (queries.length > 1) {
                    ResultMarshall rm = Query.marshallResults(queries, NetFSE.LOCAL_IP_ADDRESS,
                            NetFSE.RESULT_SERVER_PORT, query.getID());
                    this.queryServer.getResultServer().addResultMarshall(rm);
                    Vector<String> tempNames = new Vector<String>();
                    for (int i = 0; i < queryNodes.length; i++) {
                        if (queryNodes[i].getName().compareTo(NetFSE.LOCAL_HOST) != 0)
                            tempNames.addElement(queryNodes[i].getName());
                        else
                            tempNames.addElement("localhost");
                    }
                    String[] nodes = new String[tempNames.size()];
                    for (int i = 0; i < nodes.length; i++)
                        nodes[i] = (String) tempNames.elementAt(i);
                    if (nodes.length == 0) {
                        nodes = new String[1];
                        nodes[0] = "localhost";
                    }
                    for (int i = 0; i < queries.length; i++) {
                        queries[i] = queries[i].detach(queryServer.nextQueryID());
                        if (nodes[i % nodes.length].compareTo("localhost") == 0) {
                            this.process(queries[i], mdb);
                        } else {
                            SocketFactory sf2 = null;
                            if (NetFSE.USE_SSL_QUERY)
                                sf2 = SSLSocketFactory.getDefault();
                            else
                                sf2 = SocketFactory.getDefault();
                            Socket sock = null;
                            try {
                                sock = sf2.createSocket(nodes[i % nodes.length], NetFSE.QUERY_SERVER_PORT);
                            } catch (IOException ioe) {
                                System.err.println("Node '" + nodes[i % nodes.length] + "' not responding.");
                            }
                            if (sock != null) {
                                PrintWriter socketOut = new PrintWriter(sock.getOutputStream());
                                DataInputStream socketIn = new DataInputStream(sock.getInputStream());
                                socketOut.println(queries[i].toXML());
                                System.out.println("Sending query to " + nodes[i % nodes.length] + ": "
                                        + queries[i].toXML());
                                socketOut.println("<--DiSARM: end query-->");
                                socketOut.flush();
                                socketIn.readInt();
                                socketOut.close();
                                socketIn.close();
                                sock.close();

                            } else {
                                this.process(queries[i], mdb);
                            }

                        }
                    }
                    return 0;
                } else
                    query = queries[0];
            }

            // query.generateRecordFilter();
            try {
                socketResult = (SocketResult) query.getResult();
            } catch (ClassCastException cce) {
                marshalledResult = (MarshalledResult) query.getResult();
            }
            // RecordShaper recordShaper = query.getRecordShaper();
            // RecordFilter recordFilter = query.getRecordFilter();
            if (socketResult != null) {
                // System.out.println("SOCKET RESULT " + socketResult.host + ":"
                // + socketResult.port);
                resultSocket = sf.createSocket(socketResult.host, socketResult.port);
                resultSocket.setSendBufferSize(NetFSE.SOCKET_SEND_SIZE);
                resultOut = new DataOutputStream(resultSocket.getOutputStream());
            } else {
                // System.out.println("MARSHALLED RESULT " +
                // marshalledResult.host + ":" + marshalledResult.port);
                resultSocket = sf.createSocket(marshalledResult.host, marshalledResult.port);
                resultSocket.setSendBufferSize(NetFSE.SOCKET_SEND_SIZE);
                resultOut = new DataOutputStream(resultSocket.getOutputStream());
                resultOut.writeInt(marshalledResult.queryID);
                resultOut.writeInt(NetFSE.RESULT_SERVER_FLAG_DONE);
            }

            boolean done = false;
            DateRange[] dates = query.getDateRanges();
            TimeRange[] times = query.getTimeRanges();
            TimestampRange[] timestampRanges = query.getTimestampRanges();
            if (times.length > 1)
                throw (new QueryException("Support for multiple time ranges currently not implemented."));
            if (dates.length > 0) {
                Vector<DateRange> tempDateRanges = new Vector<DateRange>();
                for (int i = 0; i < dates.length; i++) {
                    nfse.query.QueryDate[] tempDates = dates[i].getDates();
                    for (int j = 0; j < tempDates.length; j++)
                        tempDateRanges.addElement(new DateRange(tempDates[j]));
                }
                dates = new DateRange[tempDateRanges.size()];
                for (int i = 0; i < dates.length; i++)
                    dates[i] = (DateRange) tempDateRanges.elementAt(i);
            }

            Criteria[] criteria = query.getCriteria();
            int[] predicates = new int[criteria.length + 1];
            String[] values1 = new String[criteria.length + 1];
            String[] values2 = new String[criteria.length + 1];
            GlobalField[] fields = new GlobalField[criteria.length + 1];
            fields[fields.length - 1] = NetFSE.getGlobalField(NetFSE.FIELD_START_TS);
            predicates[predicates.length - 1] = NetFSE.PREDICATE_RANGE;
            for (int i = 0; i < criteria.length; i++) {
                GlobalField f = NetFSE.getGlobalField(criteria[i].field);

                if (f == null) {
                    System.err.println("StaticQueryProcessor: Field was null.");
                    resultOut.close();
                    resultSocket.close();
                    return -1;
                } else
                    fields[i] = f;
                predicates[i] = criteria[i].predicate;
                if (criteria[i].predicate == NetFSE.PREDICATE_RANGE) {
                    values1[i] = NetFSEUtil.convertToHex(criteria[i].field, criteria[i].value1);
                    values2[i] = NetFSEUtil.convertToHex(criteria[i].field, criteria[i].value2);
                } else {
                    values1[i] = NetFSEUtil.convertToHex(criteria[i].field, criteria[i].value1);
                }
            }
            byte[] bytes = null;
            int count = 0;

            Class.forName("org.postgresql.Driver");
            String url = "jdbc:postgresql://localhost/" + NetFSE.getTypeName(type).toLowerCase();
            Properties props = new Properties();
            props.setProperty("user", NetFSE.DYNAMIC_DB_USER);
            props.setProperty("password", NetFSE.DYNAMIC_DB_PASSWORD);
            props.setProperty("client_encoding", "SQL_ASCII");

            Connection tempConnection = null;
            Statement tempStatement = null;
            try {
                tempConnection = DriverManager.getConnection(url, props);
            } catch (Exception e) {
                url = "jdbc:postgresql://localhost/" + NetFSE.METADATA_DATABASE;
                props.setProperty("user", NetFSE.DYNAMIC_DB_USER);
                props.setProperty("password", NetFSE.DYNAMIC_DB_PASSWORD);
                tempConnection = DriverManager.getConnection(url, props);
                tempStatement = tempConnection.createStatement();
                tempStatement.execute("create database " + NetFSE.getTypeName(type).toLowerCase() + " encoding='SQL_ASCII'");

                tempStatement.close();
                tempConnection.close();

                url = "jdbc:postgresql://localhost/" + NetFSE.getTypeName(type).toLowerCase();
                props.setProperty("user", NetFSE.DYNAMIC_DB_USER);
                props.setProperty("password", NetFSE.DYNAMIC_DB_PASSWORD);
                
                tempConnection = DriverManager.getConnection(url, props);
            }
            tempStatement = tempConnection.createStatement();
            for (int y = 0; y < timestampRanges.length; y++) {
                // System.out.println("TIMESTAMP SEARCH");
                // System.out.println(timestampRanges[y].toXML());
                Run[] runs = mdb.getRuns(timestampRanges[y].startTS, timestampRanges[y].endTS, sensor);
                for (int x = 0; x < runs.length; x++) {

                    boolean droppedTimeSearch = false;
                    if ((runs[x].getMinTime() >= timestampRanges[y].startTS.getTime() / 1000)
                            && (runs[x].getMaxTime() <= timestampRanges[y].endTS.getTime() / 1000)) {

                        droppedTimeSearch = true;
                    }
                    values1[values1.length - 1] = null;
                    values2[values2.length - 1] = null;

                    se = new SearchEngine(mdb, tempConnection, rec);
                    se.performSearch(runs[x].getMonth(), runs[x].getDay(), runs[x].getYear(), runs[x].getRun(), type,
                            sensor, fields, values1, values2, predicates);

                    String line = null;
                    NetFSERecordFactory factory = NetFSE.getRecordFactory(type);
                    NetFSERecord dummyRecord = factory.generate();
                    line = se.getNextRecord(tempStatement);
                    long trStartTS = timestampRanges[y].startTS.getTime();
                    long trEndTS = timestampRanges[y].endTS.getTime();
                    while (line != null) {
                        if (dummyRecord.hasBinary()) 
                            line = NetFSEUtil.convertFromHex(line);
                        if (droppedTimeSearch) {
                            count++;
                            bytes = line.getBytes();
                            try {
                                resultOut.writeInt(type);
                                resultOut.writeInt(sensor);
                                resultOut.writeInt(bytes.length);
                                resultOut.write(bytes);
                            } catch (IOException ioe) {
                                line = null;
                                done = true;
                            }
                        } else {
                            NetFSERecord rec = null;
                            try {
                                rec = factory.generate(line, sensor);
                            } catch (Exception e) {
                                rec = null;
                                e.printStackTrace();
                            }
                            if (rec != null) {
                                long startts = rec.getStartTS().getTime();
                                if (startts >= trStartTS && startts <= trEndTS) {
                                    count++;
                                    bytes = line.getBytes();
                                    try {
                                        resultOut.writeInt(type);
                                        resultOut.writeInt(sensor);
                                        resultOut.writeInt(bytes.length);
                                        resultOut.write(bytes);
                                    } catch (IOException ioe) {
                                        line = null;
                                        done = true;
                                    }
                                }
                            }
                        }

                        if (line != null)
                            line = se.getNextRecord(tempStatement);
                    }

                    //se.close();
                }
            }

            for (int y = 0; y < dates.length; y++) {
                if (!done) {
                    // System.out.println("DATE SEARCH");
                    Run[] runs = mdb.getRuns(dates[y].getStartDate(), dates[y].getEndDate(), sensor);
                    for (int x = 0; x < runs.length; x++)
                        System.out.println(runs[x]);
                    for (int x = 0; x < runs.length; x++) {
                        if (done)
                            break;

                        if (data.length > 1) {
                            System.err.println("Number of data types greater than 1. Not good!");
                        } else if (data.length == 0) {
                            System.err.println("Number of data types is 0. Really not good!");
                            return -1;
                        }

                        se = new SearchEngine(mdb, tempConnection, rec);
                        se.performSearch(runs[x].getMonth(), runs[x].getDay(), runs[x].getYear(), runs[x].getRun(),
                                type, sensor, fields, values1, values2, predicates);

                        String line = se.getNextRecord(tempStatement);
                        while (line != null) {
                            count++;
                            //System.out.println(line);
                            bytes = line.getBytes();
                            try {
                                resultOut.writeInt(type);
                                resultOut.writeInt(sensor);
                                resultOut.writeInt(bytes.length);
                                resultOut.write(bytes);
                            } catch (IOException ioe) {
                                line = null;
                                done = true;
                            }

                            if (line != null)
                                line = se.getNextRecord(tempStatement);
                        }
                        //se.close();

                    }
                }
            }
            tempStatement.close();
            tempConnection.close();

            if (resultOut != null) {
                if (marshalledResult != null) {
                    resultOut.writeInt(-100);
                    resultOut.writeInt(count);
                    DataInputStream resultIn = new DataInputStream(resultSocket.getInputStream());
                    resultIn.readInt();
                    resultIn.close();
                    System.out.println("StaticQueryProcessor2: finished query #" + marshalledResult.queryID
                            + ", records=" + count);
                }
                resultOut.close();
                resultSocket.close();
            }
            if (statDB != null)
                statDB.closeConnection();
        } catch (Exception E) {
            E.printStackTrace();
            System.out.println("Client closed connection; cleaning up.");
            try {
                if (se != null) {
                    //se.close();
                }
                if (socketResult != null) {
                    resultSocket = sf.createSocket(socketResult.host, socketResult.port);
                    resultOut = new DataOutputStream(resultSocket.getOutputStream());
                } else if (marshalledResult != null) {
                    resultSocket = sf.createSocket(marshalledResult.host, marshalledResult.port);
                    resultOut = new DataOutputStream(resultSocket.getOutputStream());
                    resultOut.writeInt(marshalledResult.queryID);
                    resultOut.writeInt(NetFSE.RESULT_SERVER_FLAG_DONE);
                    resultOut.writeInt(-100);
                    resultOut.writeInt(0);
                    DataInputStream resultIn = new DataInputStream(resultSocket.getInputStream());
                    resultIn.readInt();
                    resultIn.close();
                }
                if (resultOut != null) {
                    resultOut.close();
                    resultSocket.close();
                }

            } catch (Exception e2) {
                e2.printStackTrace();
            }
        }

        try {
            if (ss != null)
                ss.close();
            if (resultSocket != null)
                resultSocket.close();
        } catch (IOException ioe) {
            ioe.printStackTrace();
        }
        return 0;
    }

}